package com.hujf.flink.tableSql.sql;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @Author hujf
 * @Date 2022/8/26 16:03
 * @Description TODO
 */
public class FlinkSqlTest {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment();
        environment.setParallelism(1);

        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(environment);

// 对已注册的表进行 SQL 查询
// 注册名为 “Orders” 的表
//        tableEnv.executeSql("CREATE TABLE Orders (`id` INT) WITH (" +
//                "'connector.type' = 'jdbc',"+
//                "'connector.url' = 'jdbc:mysql://10.0.6.83:3306/daas_pcdb02',"+
//                "'connector.username'='root',"+
//                "'connector.password'='mysql',"+
//                "'connector.table'='view_focal_point_notice_data')");
//// 在表上执行 SQL 查询，并把得到的结果作为一个新的表
//        Table result = tableEnv.sqlQuery(
//                "SELECT * FROM Orders");
//
//        tableEnv.createTemporaryView("results" ,result);
//        tableEnv.toDataStream(result).print();



        tableEnv.executeSql("CREATE TABLE common_company_portrait (`company_id` STRING, company_name STRING) WITH (" +
                "'connector' = 'elasticsearch-6',"+
                "'host'='http://10.0.6.33:9100/',"+
                "'index'='common_company_portrait')");
// 在表上执行 SQL 查询，并把得到的结果作为一个新的表
        Table resultEs = tableEnv.sqlQuery(
                "SELECT* FROM common_company_portrait limit 10");



        tableEnv.toDataStream(resultEs).print();





        environment.execute();

    }
}
